package com.demo;


import com.alibaba.fastjson.JSON;
import com.demo.model.PsdkLogAppUser;
import com.demo.model.PsdkTradeInfo;
import com.demo.model.UsdkTrade;
import com.demo.model.UsdkUserLoginLog;
import java.util.Map;
import java.util.Properties;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

public class TableRegister {
    // 维表可以按需求改成mysql读
    // 参考：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/jdbc.html
    public static void registerDim(StreamTableEnvironment tableEnv, String zkQuorum, String znodeParent) {
        tableEnv.executeSql("create table currency_rate ( "
                + "currency STRING, "
                + "cf ROW<er STRING>, "
                + "PRIMARY KEY (currency) NOT ENFORCED "
                // WATERMARK FOR ts_field AS ts_field - INTERVAL '2' SECOND
                + ") "
                + " WITH ( "
                + " 'connector' = 'hbase-1.4', "
                + " 'table-name' = 'exchange_rate', "
                + " 'zookeeper.quorum' = '" + zkQuorum + "', "
                + " 'zookeeper.znode.parent' = '" + znodeParent + "' "
                + ")");

        tableEnv.executeSql("create table channel_info ( "
                + "channel_id STRING, "
                + "cf ROW<ch STRING>, "
                + "PRIMARY KEY (channel_id) NOT ENFORCED "
                + ") "
                + " WITH ( "
                + " 'connector' = 'hbase-1.4', "
                + " 'table-name' = 'channel_info', "
                + " 'zookeeper.quorum' = '" + zkQuorum + "', "
                + " 'zookeeper.znode.parent' = '" + znodeParent + "' "
                + ")");

        tableEnv.executeSql("create table pub_app_mapping ( "
                + "app_id STRING, "
                + "cf ROW<g_name STRING, pub_id STRING, os STRING, start_date STRING>, "
                + "PRIMARY KEY (app_id) NOT ENFORCED "
                // WATERMARK FOR ts_field AS ts_field - INTERVAL '2' SECOND
                + ") "
                + " WITH ( "
                + " 'connector' = 'hbase-1.4', "
                + " 'table-name' = 'pub_app_mapping', "
                + " 'zookeeper.quorum' = '" + zkQuorum + "', "
                + " 'zookeeper.znode.parent' = '" + znodeParent + "' "
                + ")");
    }

    public static void registerKafkaTable(StreamTableEnvironment tableEnv, String server, String groupId, String topic) {
        tableEnv.connect(
                new Kafka()
                        .version("universal")
                        .topic(topic)
                        .startFromEarliest()
//                        .startFromLatest()
//                        .startFromEarliest()
//                        .property("zookeeper.connect", "localhost:2181")
//                        .property("scan.startup.mode", "latest-offset")
                        .property("bootstrap.servers", server)
                        .property("group.id", groupId)
        ).withFormat(
                new Json()
        ).withSchema(
                new Schema()
                        .field("app_id", DataTypes.STRING())
                        .field("complete_time",DataTypes.STRING())
                        .field("audit_update",DataTypes.STRING())
                        .field("added_time",DataTypes.STRING())
                        .field("@timestamp",DataTypes.STRING())
                        .field("channel_id",DataTypes.STRING())
                        .field("@version",DataTypes.STRING())
                        .field("state",DataTypes.STRING())
                        .field("fee",DataTypes.STRING())
                        .field("user_id",DataTypes.STRING())
                        .field("audit_create",DataTypes.STRING())
                        .field("pay_type",DataTypes.STRING())
                        .field("currency",DataTypes.STRING())
                        .field("type",DataTypes.STRING())
                        .field("id",DataTypes.STRING())
                        .field("login_time",DataTypes.STRING())
                        .field("sdk_user_id",DataTypes.STRING())
                        .field("status",DataTypes.STRING())
                        .field("create_time",DataTypes.STRING())
                        .field("sdk_id",DataTypes.STRING())
                        .field("trade_id",DataTypes.STRING())
                        .field("log_id",DataTypes.STRING())
                        .field("proctime", DataTypes.TIMESTAMP()).proctime()

        ).inAppendMode().createTemporaryTable("kafkaSourceTable");
    }

  /**
   * 暂不使用
   * @param tableEnv
   * @param kafkaStream
   */
    @Deprecated
    public static void registerKafkaTable(StreamTableEnvironment tableEnv, DataStream<String> kafkaStream) {
        //psdk_trade_info


        DataStream<PsdkTradeInfo> psdkTradeInfoDataStream = kafkaStream
                .map(new MapFunction<String, PsdkTradeInfo>() {
                    private static final long serialVersionUID = 1471936326697828381L;

                    @Override
                    public PsdkTradeInfo map(String value) {
                        if ("psdk_trade_info".equals(JSON.parseObject(value, Map.class).get("type"))) {
                            return JSON.parseObject(value, PsdkTradeInfo.class);
                        } else {
                            return null;
                        }
                    }
                }).filter(new FilterFunction<PsdkTradeInfo>() {
                    @Override
                    public boolean filter(PsdkTradeInfo psdkTradeInfo) throws Exception {
                        System.out.println(psdkTradeInfo);
                        return psdkTradeInfo == null;
                    }
                });

        //psdk_log_app_user
        DataStream<PsdkLogAppUser> psdkLogAppUserDataStream = kafkaStream
                .map(new MapFunction<String, PsdkLogAppUser>() {
                    private static final long serialVersionUID = 1471936326697828381L;

                    @Override
                    public PsdkLogAppUser map(String value) {
                        if ("psdk_log_app_user".equals(JSON.parseObject(value, Map.class).get("type"))) {
                            return JSON.parseObject(value, PsdkLogAppUser.class);
                        } else {
                            return null;
                        }
                    }
                }).filter(new FilterFunction<PsdkLogAppUser>() {
                    @Override
                    public boolean filter(PsdkLogAppUser psdkLogAppUser) throws Exception {
                        return psdkLogAppUser != null;
                    }
                });

        //usdk_trade
        DataStream<UsdkTrade> usdkTradeDataStream = kafkaStream
                .map(new MapFunction<String, UsdkTrade>() {
                    private static final long serialVersionUID = 1471936326697828381L;

                    @Override
                    public UsdkTrade map(String value) {
                        if ("usdk_trade".equals(JSON.parseObject(value, Map.class).get("type"))) {
                            return JSON.parseObject(value, UsdkTrade.class);
                        } else {
                            return null;
                        }
                    }
                }).filter(new FilterFunction<UsdkTrade>() {
                    @Override
                    public boolean filter(UsdkTrade usk) throws Exception {
                        return usk != null;
                    }
                });

        //usdk_user_login_log
        DataStream<UsdkUserLoginLog> usdkUserLoginLogDataStream = kafkaStream
                .map(new MapFunction<String, UsdkUserLoginLog>() {
                    private static final long serialVersionUID = 1471936326697828381L;

                    @Override
                    public UsdkUserLoginLog map(String value) {
                        if ("usdk_user_login_log".equals(JSON.parseObject(value, Map.class).get("type"))) {
                            return JSON.parseObject(value, UsdkUserLoginLog.class);
                        } else {
                            return null;
                        }
                    }
                }).filter(new FilterFunction<UsdkUserLoginLog>() {
                    @Override
                    public boolean filter(UsdkUserLoginLog usdkUserLoginLog) throws Exception {
                        return usdkUserLoginLog != null;
                    }
                });

        tableEnv.createTemporaryView("psdk_trade_info", psdkTradeInfoDataStream);
        tableEnv.createTemporaryView("psdk_log_app_user", psdkLogAppUserDataStream);
        tableEnv.createTemporaryView("usdk_trade", usdkTradeDataStream);
        tableEnv.createTemporaryView("usdk_user_login_log", usdkUserLoginLogDataStream);
    }

    private static Properties kafkaProperties(String server, String groupId) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", server);
        properties.put("group.id", groupId);
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return properties;
    }
}
